package com.bbx.flink.demo.windowfuncation;

import com.esotericsoftware.minlog.Log;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.CharSet;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.nio.charset.Charset;

@Slf4j
public class AverageAggregateFuncatinonDemo {
    public static void  main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("114.116.104.74",9900)
                .map((MapFunction<String ,Integer>)value ->Integer.parseInt(value))
                .keyBy(i->i%2)
                .timeWindow(Time.seconds(60))
                .aggregate(new AggregateFunction<Integer, Double, String>() {
                    @Override
                    public Double createAccumulator() {
                        log.info("初始化累加器");
                        return 0.00;
                    }

                    @Override
                    public Double add(Integer integer, Double aDouble) {
                        log.info("执行Add 方法");
                        return new Double(integer)+aDouble;
                    }

                    @Override
                    public String getResult(Double aDouble) {
                        log.info("获取最后结果");
                        return String.valueOf(aDouble);
                    }

                    @Override
                    public Double merge(Double aDouble, Double acc1) {
                        log.info("合并数据");
                        return aDouble+acc1;
                    }
                }).print()
                ;

        env.execute();



    }



}

